package com.ibm.datamodel.timeseries.parser;

import java.util.Arrays;

import com.fasterxml.jackson.databind.JsonNode;
import com.fasterxml.jackson.databind.node.ArrayNode;
import com.fasterxml.jackson.databind.node.ObjectNode;
import com.ibm.cps.message.MessageFields;
import com.ibm.cps.message.ParsedDataMessage;
import com.ibm.cps.message.ParsedDataMessageBuilder;
import com.ibm.cps.message.RawDataMessage;
import com.ibm.interfaces.ITimeSeriesSchema;
import com.ibm.util.DataTypes;
import com.ibm.util.EntryPair;
import com.ibm.util.ErrorCode;
import com.ibm.util.JSONUtility;
import com.ibm.util.exception.CPSException;

/**
 * @author Yaoliang Chen
 * @version Nov 20, 2015 TODO P2 Performance Optimization for tskey extraction
 */
public class JsonParser extends ProtostuffMessageParser {

	public JsonParser(ITimeSeriesSchema schema) throws CPSException {
		super(schema);
	}

	/**
	 * TODO performance optimization
	 */
	@Override
	public Comparable getTimeSeriesKey(byte[] data) throws CPSException {
		JsonNode node = JSONUtility.fromString(new String(data));
		if (node.isArray())
			return node.get(schema.getKeyIndex()).asLong(1);
		return node.get(schema.getKeyName()).asLong(1);
	}

	@Override
	public ParsedDataMessage parseDataMessage(RawDataMessage jsonMsg) throws CPSException {
		ParsedDataMessage message = schema.newMessage();
		String jsonStr = new String(jsonMsg.getMessageBytes());

		try {
//          it cannot merge custom datatype or array.rewrite it.
//			JsonIOUtil.mergeFrom(jsonMsg.getMessageBytes(), message, this, false);
			JsonNode node  = JSONUtility.fromStringWithNoException(jsonStr);
			ParsedDataMessageBuilder builder = ParsedDataMessageBuilder.createBasicBuilder(schema, 
					jsonMsg.getTsKey(), node.get(schema.getTagNames()[schema.getTimestampIndex()]).asLong());

			// iterator the schema datatype
			String[] tags = schema.getTagNames();
			int tagCount = schema.getTagCount();
			for (int i = 0; i < tagCount; i++) {
				String field = tags[i];
				if (i == schema.getTimestampIndex() || i == schema.getKeyIndex()) {
					// skip the timestamp and tskey,finish already.
					continue;
				}
				switch (schema.getDataTypes()[i]) {
					case DataTypes.DT_BOOLEAN:
						builder.append(field, node.get(field).asBoolean());
						break;
					case DataTypes.DT_DOUBLE:
						builder.append(field, node.get(field).asDouble());
						break;
					case DataTypes.DT_DOUBLE_ARRAY:
						ArrayNode doubleNode = (ArrayNode) node.get(field);
						Double[] doubleArray = new Double[doubleNode.size()];
						for (int j = 0; j < doubleArray.length; j++) {
							doubleArray[j] = doubleNode.get(i).asDouble();
						}
						builder.append(field, doubleArray);
						break;
					case DataTypes.DT_INT_ARRAY:
						ArrayNode intNode = (ArrayNode) node.get(field);
						Integer[] intArray = new Integer[intNode.size()];
						for (int j = 0; j < intArray.length; j++) {
							intArray[j] = intNode.get(i).asInt();
						}
						builder.append(field, intArray);
						break;
					case DataTypes.DT_LONG:
						builder.append(field, node.get(field).asLong());
						break;
					case DataTypes.DT_STRING:
						builder.append(field, node.get(field).asText());
						break;
					case DataTypes.DT_STRING_ARRAY:
						ArrayNode strNode = (ArrayNode) node.get(field);
						String[] strArray = new String[strNode.size()];
						for (int j = 0; j < strArray.length; j++) {
							strArray[j] = strNode.get(i).asText();
						}
						builder.append(field, strArray);
						break;
					case DataTypes.DT_WILDCARD:
						EntryPair entryPair = new EntryPair(field, jsonStr);
						builder.append(field, entryPair);
						break;
					default:
						throw new CPSException(ErrorCode.NO_IMPLEMENTATION_CODE, 
								"unsuport data type for tagname["+field+"] datatype[" + 
								schema.getDataTypes()[i]+"]");
				}
				
			}
			
			message = builder.toParsedDataMessage();
		} catch (Exception e) {
			System.out.println("checking in JsonParser" + jsonStr);
			throw new CPSException(ErrorCode.PARSE_ERROR_CODE, "error while parse the message; " + e.toString());
		}
		return message;
	}

	@Override
	public byte[] restoreFromDataMessage(ParsedDataMessage message) throws CPSException {
//      to support array datatype and wildcard,rewrite
//		return JsonIOUtil.toByteArray(message, this, false);
		ObjectNode node = JSONUtility.newObjectNode();
		try {

			// iterator the schema datatype
			String[] tags = schema.getTagNames();
			int tagCount = schema.getTagCount();
			for (int i = 0; i < tagCount; i++) {
				String field = tags[i];
				switch (schema.getDataTypes()[i]) {
				case DataTypes.DT_BOOLEAN:
				case DataTypes.DT_DOUBLE:
				case DataTypes.DT_LONG:
				case DataTypes.DT_STRING:
					node.put(field, message.getValue(i).toString());
					break;
				case DataTypes.DT_DOUBLE_ARRAY:
				case DataTypes.DT_INT_ARRAY:
				case DataTypes.DT_STRING_ARRAY:
					Object[] values = message.getArrayValue(i);
					node.put(field, Arrays.toString(values));
					break;
				case DataTypes.DT_WILDCARD:
					EntryPair value = (EntryPair) message.getValue(i);
					node.put(field, value.overview());
					break;
				default:
					throw new CPSException(ErrorCode.NO_IMPLEMENTATION_CODE, 
							"unsuport data type for tagname["+field+"] datatype[" + 
							schema.getDataTypes()[i]+"]");
				}
				
			}
		} catch (Exception e) {
			throw new CPSException(ErrorCode.PARSE_ERROR_CODE, "error while parse the message; " + e.getMessage());
		}
		return node.toString().getBytes();
	
	}

}
